package com.zhc.tools.core.fileimport.strategy;

import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.Lists;
import com.zhc.tools.core.fileimport.IProcessor;
import com.zhc.tools.core.fileimport.StrategyContext;
import com.zhc.tools.core.fileimport.domain.ProcessorMethodEnum;

/**
 * 说明：批量导入文件多线程处理类
 * @author zhc
 */
public class MultiThreadProcessor implements Runnable, IProcessor {
	
	private final Log logger = LogFactory.getLog(this.getClass());
	
	protected StrategyContext strategyContext;
	
	/**
	 * 要处理的数据
	 */
	protected List<?> originDatas;
	
	/**
	 * 多线程等待
	 */
	private CountDownLatch countDownLatch;
	
	/**
	 * 每个线程执行条数
	 */
	private int fetchSize;
	
	/**
	 * 代理具体处理器
	 */
	private IProcessor targetProcessor;
	
	/**
	 * 具体处理方法枚举
	 */
	private ProcessorMethodEnum processorMethod;
	
	/**
	 * 单条抓取任务锁
	 */
	Lock lockFetchData = new ReentrantLock();
	/**
	 * 多条抓取任务锁
	 */
	Lock lockFetchMutilData = new ReentrantLock();
	
	public MultiThreadProcessor(IProcessor targetProcessor) {
		this.targetProcessor = targetProcessor;
	}
	
	@Override
	public boolean doExecute(StrategyContext strategyContext, List<?> originDatas, ProcessorMethodEnum processorMethod) {
		this.strategyContext = strategyContext;
		this.originDatas = originDatas;
		this.processorMethod = processorMethod;
		return this.initThread();
	}
	
	private boolean initThread() {
		long startTime = System.currentTimeMillis();
		// 启动线程数量
		int threadNum = this.strategyContext.getFileStrategySetting().getThreadNum();
		int maxThreadNum = Runtime.getRuntime().availableProcessors() * 4;
		if (threadNum <= 0 || threadNum > maxThreadNum) {
			threadNum = maxThreadNum/4;
		}
		
		int totalSize = this.originDatas.size();
		
		int useMultiNumMin = strategyContext.getFileStrategySetting().getUseMultiNumMin();
		// 当数据数量大于 useMultiNumMin （默认50）时，才使用多线程，否则只创建一个线程
		if (useMultiNumMin >= totalSize) {
			threadNum = 1;
		}
		
		// 每次抓取数量
		this.fetchSize = totalSize%threadNum==0 ? totalSize/threadNum : (totalSize/threadNum+1);
		
		int actualThreadNum = totalSize%fetchSize == 0 ? totalSize/fetchSize : (totalSize/fetchSize + 1);
		if (actualThreadNum < threadNum) {
			threadNum = actualThreadNum;
		}
		
		if (this.fetchSize < threadNum * 10) {
			this.logger.warn("参数设置不合理，系统性能不佳。【每次从数据库获取的数量fetchSize】 >= 【线程数量threadNum】 *【最少循环次数10】 ");
        }
		this.countDownLatch = new CountDownLatch(threadNum);
		for (int i = 0; i < threadNum; i++) {
            this.startThread(i);
        }
		
		try {  
            //调用await方法阻塞当前线程，等待子线程完成后在继续执行
            this.countDownLatch.await();
        } catch (InterruptedException e) {  
            this.logger.debug("MultiThreadProcessor#initThread await error:", e);
        }
		long endTime = System.currentTimeMillis();
		
		if (this.logger.isDebugEnabled()) {
			this.logger.debug(String.format("MultiThreadProcessor#initThread processorMethod【%s】long time 【%s】ms dataSize【%s】 ", 
					processorMethod, (endTime - startTime), totalSize));
		} else if (this.logger.isInfoEnabled()) {
			this.logger.info(String.format("MultiThreadProcessor#initThread processorMethod【%s】long time 【%s】ms dataSize【%s】 OK ", 
					processorMethod, (endTime - startTime), totalSize));
		}
		return true;
	}
	
	private void startThread(int index) {
        Thread thread = new Thread(this);
        String threadName = this.strategyContext.getBatch() + "-" + this.processorMethod + "-exe" + index;
        thread.setName(threadName);
        thread.start();
    }
	
	/**
     * 获取单个任务，注意lock是必须
     * 
     * @return
     */
    public Object getData() {
    	lockFetchData.lock();
        try {
        	Object result = null;
            while (true) {
                if (this.originDatas.size() > 0) {
                    result = this.originDatas.remove(0); // 按正序处理
                } else {
                    return null;
                }
                return result;
            }
        } finally {
        	lockFetchData.unlock();
        }
    }
	
	/**
     * 获取多个任务，注意lock是必须
     * 
     * @return
     */
	public List<Object> getMutilData() {
        lockFetchMutilData.lock();
        try {
            if (this.originDatas.size() == 0) {
                return null;
            }
            
            int size = this.originDatas.size() > fetchSize ? fetchSize : this.originDatas.size();

            List<Object> result = Lists.newArrayList();
            int point = 0;
            Object tmpObject = null;
            while (point < size && ((tmpObject = this.getData()) != null)) {
                result.add(tmpObject);
                point = point + 1;
            }
            if (result.size() == 0) {
                return null;
            } else {
                return result;
            }
        } finally {
        	lockFetchMutilData.unlock();
        }
    }

	@Override
	public void run() {
		long startTime = System.currentTimeMillis();
		try {
			boolean flag = false;
			List<Object> executeData = getMutilData();
			flag = this.targetProcessor.doExecute(strategyContext, executeData, processorMethod);
			long endTime = System.currentTimeMillis();
			if (this.logger.isDebugEnabled()) {
				this.logger.debug(String.format("MultiThreadProcessor#run result【%s】 processorMethod【%s】long time 【%s】ms dataSize【%s】 datas【%s】 ", 
						flag, processorMethod, (endTime - startTime), executeData.size(), JSON.toJSONString(executeData)));
			} else if (this.logger.isInfoEnabled()) {
				this.logger.info(String.format("MultiThreadProcessor#run result【%s】 processorMethod【%s】long time 【%s】ms dataSize【%s】 OK ", 
						flag, processorMethod, (endTime - startTime), executeData.size()));
			}
		} catch (Exception e) {
			this.logger.error("MultiThreadProcessor#run ERROR:", e);
		} finally {
			this.countDownLatch.countDown();
		}
	}
}
